package com.zhaojiyuan.boot.base.common.util;


import java.util.ArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
import lombok.extern.slf4j.Slf4j;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;

/**
 * JAVA多线程事务
 * <p>1.addFunction 添加要异步执行的方法
 * <p>2.execute方法中，使用全局的计数器和异常标记字段，统计个异步线程执行的结果 当所有异步线程执行完之后，根据异常标记字段判断是回滚还是提交事务。
 *
 * @author zhaojiyuan
 */
@Slf4j
public class MultiThreadTransactionComponent {

  /**
   * 线程警告数警告日志
   */
  private final int THREAD_COUNT_WARN = 5;

  /**
   * 事务管理器
   */
  private PlatformTransactionManager platformTransactionManager;

  /**
   * 线程池
   */
  private ExecutorService executorService;

  /**
   * 有序 任务 集合
   */
  private ArrayList<Supplier> supplierList = new ArrayList<>(16);

  /**
   * 创建执行计数器
   */
  private CountDownLatch countDownLatch;

  /**
   * 异常标志位
   */
  private AtomicReference<Boolean> isError = new AtomicReference<>(false);

  /**
   * 构造函数
   *
   * @param transactionManager 事务管理器
   */
  public MultiThreadTransactionComponent(PlatformTransactionManager transactionManager) {
    this.platformTransactionManager = transactionManager;
  }

  /**
   * 添加要异步执行的方法程序
   *
   * @param supplier 任务
   */
  public void addFunction(Supplier supplier) {
    supplierList.add(supplier);
  }

  /**
   * 执行队列中的任务
   */
  public Boolean execute() {
    if (supplierList.size() > THREAD_COUNT_WARN) {
      log.warn("当前线程数为：{}个，线程数大于数据库连接数将发生死锁");
    }
    // 创建一个线程池
    executorService = Executors.newFixedThreadPool(supplierList.size());
    // 初始化 线程 计数器
    countDownLatch = new CountDownLatch(supplierList.size());
    // 提交任务
    supplierList.forEach(it -> executorService.submit(new TransactionRunnable(platformTransactionManager, it)));

    try {
      // 主线程 等待计数器为 0 时 进行提交或回滚
      countDownLatch.await();
      if (isError.get()) {
        log.error("多线程执行失败");
        return false;
      }
    } catch (InterruptedException e) {
      // 打印异常日志
      log.error("多线程执行失败" + e.getMessage());
      return false;
    }
    return true;
  }


  /**
   * 实现 Runnable 接口
   */
  class TransactionRunnable implements Runnable {

    /**
     * 事务管理器
     */
    private PlatformTransactionManager platformTransactionManager;
    /**
     * 当前任务
     */
    private Supplier supplier;

    /**
     * 构造函数
     *
     * @param platformTransactionManager 事务管理器
     * @param supplier                   当前任务
     */
    public TransactionRunnable(PlatformTransactionManager platformTransactionManager, Supplier supplier) {
      this.platformTransactionManager = platformTransactionManager;
      this.supplier = supplier;
    }

    @Override
    public void run() {
      log.debug("子线程执行：开始");
      DefaultTransactionDefinition def = new DefaultTransactionDefinition();
      // 创建当前线任务的事务
      def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
      TransactionStatus transaction = this.platformTransactionManager.getTransaction(def);
      log.debug("子线程执行：开始");
      try {
        // 尝试获取任务值
        supplier.get();
      } catch (Exception e) {
        // 异常时，设置错误标记
        isError.set(true);
        log.error("子线程执行：异常{}", e.getMessage());
      }
      // 线程计数器 -1
      countDownLatch.countDown();
      try {
        // 子线程 等待计数器为 0 时 进行提交或回滚
        countDownLatch.await();
        if (isError.get()) {
          //事务回滚
          platformTransactionManager.rollback(transaction);
          log.info("子线程执行：事务回滚");
        } else {
          //事务提交
          platformTransactionManager.commit(transaction);
          log.info("子线程执行：事务提交");
        }
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
  }
}

